/**
 * 
 */
package com.ffcs.crmd.tsp.server.job.selecter;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.collections.CollectionUtils;

import com.ffcs.crmd.tsp.server.api.ISelectSendChannel;

import io.netty.channel.Channel;

/**
 * 功能说明:轮询方式选择发送
 *
 * @author ZHONGFUHUA
 * 
 * @since 2.0.0-SNAPSHOT
 *
 */
public class SelectPollingImpl implements ISelectSendChannel<SelectPollingParam> {
    
    private final static Map<String, AtomicLong> jobSendCounterMap = new HashMap<String, AtomicLong>();
    
    @Override
    public Channel select(SelectPollingParam selectPollingParam) {
        List<Channel> alreadyChannelList = selectPollingParam.getAlreadyChannelList();
        List<Channel> channelRefJobList = selectPollingParam.getChannelRefJob();
        AtomicLong atomicLong = jobSendCounterMap.get(selectPollingParam.getJobName());
        if (atomicLong == null) {
            synchronized (this) {
                atomicLong = jobSendCounterMap.get(selectPollingParam.getJobName());
                if (atomicLong == null) {
                    atomicLong = new AtomicLong(1);
                }
                jobSendCounterMap.put(selectPollingParam.getJobName(), atomicLong);
            }
        }
        
        if (CollectionUtils.isNotEmpty(alreadyChannelList)) {
            channelRefJobList.removeAll(alreadyChannelList);
        }
        
        if (CollectionUtils.isNotEmpty(channelRefJobList)) {
            long index = Math.abs(atomicLong.get());
            int sindex = (int) (index % channelRefJobList.size());
            Channel channel = channelRefJobList.get(sindex);
            if (channel.isActive()) {
                atomicLong.incrementAndGet();
                return channel;
            }
        }
        return null;
    }
    
}
